package org.wyz.mapreduce.kmeans;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.ArrayList;
import java.util.List;

public class KMeansMapper extends Mapper<LongWritable, Text, Text, PointWritable> {

    private static List<List<Object>> centers = null;

    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, PointWritable>.Context context) throws IOException, InterruptedException {
        // 先根据缓存中的center表，得出当前point属于哪个簇
        if (centers == null) {
            centers = new ArrayList<>();

            URI uri = context.getCacheFiles()[0];
            String path = uri.getPath();
            BufferedReader reader = new BufferedReader(new InputStreamReader(new FileInputStream(path)));

            String line = reader.readLine();
            while (line != null) {
                List<Object> center = new ArrayList<>();
                String[] tagAndCenter = line.split("\\s");
                String[] xAndY = tagAndCenter[1].split(",");
                center.add(tagAndCenter[0]);
                center.add(Double.parseDouble(xAndY[0]));
                center.add(Double.parseDouble(xAndY[1]));
                centers.add(center);

                line = reader.readLine();
            }
        }

        String[] xAndY = value.toString().split(" ")[0].split(",");
        double x = Double.parseDouble(xAndY[0]);
        double y = Double.parseDouble(xAndY[1]);

        String tag = null;
        double minDistance = Double.MAX_VALUE;
        for (List<Object> center : centers) {
            double distance = Math.pow((double) center.get(1) - x, 2) + Math.pow((double) center.get(2) - y, 2);
            if (distance < minDistance) {
                tag = (String) center.get(0);
                minDistance = distance;
            }
        }
        context.write(new Text(tag), new PointWritable(x, y));
    }
}
